package com.cttic.kafka2hbase.server;


import cn.hutool.json.JSONObject;
import com.cttic.kafka2hbase.config.HBaseUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.Set;
import java.util.stream.Collectors;

import static com.cttic.kafka2hbase.config.KafkaConfig.PIPE_TOPIC;

@Component
public class Kafka2HBase {

    private Logger logger = LoggerFactory.getLogger(Kafka2HBase.class);


    private final static String tableName = "data_java";

    private final static String columnFamily = "message";


    @Autowired
    HBaseUtils hBaseUtils;


    /*public void test(){
        try {
            hBaseUtils.createTable("data_java","message");
        } catch (IOException e) {
            e.printStackTrace();
        }
    }*/

//    @KafkaListener(topics = PIPE_TOPIC,concurrency = "14")
    public void handle(List<ConsumerRecord<Integer,String>> records){

        logger.info("拉取了{}条",records.size());
        records.forEach( record -> {
            JSONObject jsonObject = new JSONObject(record.value());
            String positiontime = jsonObject.getStr("positiontime").substring(0, 10);
            List<String> strings = jsonObject.keySet().stream().collect(Collectors.toList());
            List<String> values = jsonObject.values().stream().map(x -> x.toString()).collect(Collectors.toList());
            try {
                hBaseUtils.insertRecords(tableName,positiontime,columnFamily,strings,values);
            } catch (IOException e) {
                e.printStackTrace();
            }
        });


    }


}
